package com.shujia.spark.mllib

import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  *
  *

我爱数加学院数加学院
因为看见，所以相信
梦想从这里开始数加学院

将中文文本数据转换成向量

1、对数据进行分词
我 爱 数加学院 数加学院
因为 看见，所以 相信
梦想 从 这里 开始 数加学院


2、构建词典向量
0  1    2      3    4    5    6    7   8   9   10
我 爱 数加学院 因为 看见  所以 相信  梦想 从 这里 开始


3、通过词典向量表示数据
1 1 2 0 0 0 0 0 0 0 0 0
0 0 0 1 1 1 1 0 0 0 0 0
0 0 1 0 0 0 0 1 1 1 1 1


稀疏向量
(11,(0,1,2),(1,1,2))
(11,(3,4,5,6),(1,1,1,1))
(11,(2,7,8,9,10),(1,1,1,1,1))



  *
  */
object TF_IDF {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local").setAppName("tf-idf")

    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)

    import sQLContext.implicits._


    val rdd = sc.textFile("spark/data/word.txt")

    //分词
    val wordDF = rdd.map(line => {
      val words = IK.fit(line)
      (line, words.mkString(" "))
    }).toDF("text", "words")


    //Tokenizer  英文分词器
    val tok = new Tokenizer()
      .setInputCol("words")
      .setOutputCol("feature")


    val fratrueDF = tok.transform(wordDF)
    fratrueDF.show()

    fratrueDF.printSchema()


    //计算tf
    val tf = new HashingTF()
      .setInputCol("feature")
      .setOutputCol("tf")
    //.setNumFeatures(1000) //太小可能导致碰撞

    val tfDF = tf.transform(fratrueDF)

    tfDF.show(false)


    ///计算idf
    val idf = new IDF()
      .setInputCol("tf")
      .setOutputCol("tf-idf")

    val model = idf.fit(tfDF)

    val idfDF = model.transform(tfDF)

    idfDF.show(false)


  }
}
